package producer;

import bean.orderbook.MatchEvent;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import thirdpart.hq.MatchData;

import java.util.Properties;

public class ExchangeProducer {
    private volatile static Producer producer;

    private ExchangeProducer() {

    }
    public static Producer getProducer(){
        if (producer ==null){
            synchronized(ExchangeProducer.class){
                if (producer==null){
                    init();
                }
            }
        }
        return producer;
    }

    public static void init(){
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.26.130:9092");
        props.put("acks", "all");// ACK应答
        props.put("retries", 1);//重试次数
        props.put("batch.size", 16384);//批次大小
        props.put("linger.ms", 1);//等待时间
        props.put("buffer.memory", 33554432);//RecordAccumulator缓冲区大小
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        producer = new KafkaProducer<>(props);
    }

    public void send(MatchEvent me, MatchData md){

        producer.send(new ProducerRecord<String, String>("flink-test",
                me.toString(),
                md.toString()));
    }


}
